package com.karaoke.android.data.udp

import com.alibaba.fastjson.JSON
import com.karaoke.android.data.domain.UdpMessage
import com.karaoke.android.data.onIO
import com.karaoke.android.data.onUI
import io.netty.buffer.Unpooled
import io.netty.channel.socket.DatagramPacket
import io.reactivex.netty.RxNetty
import io.reactivex.netty.channel.ObservableConnection
import org.simple.eventbus.EventBus
import rx.Observable
import rx.lang.kotlin.onErrorReturnNull
import java.net.InetAddress
import java.net.InetSocketAddress
import java.nio.charset.Charset


/**
 * Created by allen on 2017/5/27.
 */

class UDPClient(private val address: String)
{

    private var serverConnect: ObservableConnection<DatagramPacket, DatagramPacket>? = null
    private val serverAddress by lazy {
        InetAddress.getByName(address)
    }

    fun onConnect()
    {
        RxNetty.createUdpServer(0, {
            serverConnect = it
            it.input.flatMap { onReceiveMessage(it) }.onErrorReturnNull()
        }).withErrorHandler { onError(it) }.start()
    }

    fun onReceiveMessage(data: DatagramPacket): Observable<Void>
    {
        try
        {
            val message = JSON.parseObject(data.content().toString(Charset.defaultCharset()), UdpMessage::class.java)
            EventBus.getDefault().post(message, message.cmd)
        }
        catch (e: Exception)
        {
            e.printStackTrace()
        }

        return Observable.never()
    }

    fun sendMessage(message: UdpMessage)
    {
        Observable.unsafeCreate<Unit> {
            val value = JSON.toJSONString(message)
            val packet = DatagramPacket(Unpooled.buffer().writeBytes(value.toByteArray()), InetSocketAddress(serverAddress, 8088))
            serverConnect?.writeAndFlush(packet)
        }.onIO().onUI({}, { it.printStackTrace() })

    }

    fun onError(error: Throwable): Observable<Void>
    {
        error.printStackTrace()
        return Observable.never()
    }
}
